package com.ziroom.crm.mq.test;

import com.alibaba.fastjson.JSON;
import com.ziroom.crm.mq.rabbit.spring.core.ReliableRabbitTemplate;
import com.ziroom.crm.mq.rabbit.spring.data.RabbitMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.stream.IntStream;

@Component
@Slf4j
public class PublishDemo implements InitializingBean {

    private final ReliableRabbitTemplate amqpTemplate;

    ExecutorService executorService = Executors.newSingleThreadExecutor();

    // AtomicInteger success = new AtomicInteger();
    // AtomicInteger failed = new AtomicInteger();

    public static Set<String> successSet = new CopyOnWriteArraySet<>();
    public static Set<String> failSet = new CopyOnWriteArraySet<>();
    public static Set<String> set = new CopyOnWriteArraySet<>();

    public static String failFlag = "shouldFail";

    Random random = new Random();

    public PublishDemo(ReliableRabbitTemplate amqpTemplate) {
        this.amqpTemplate = amqpTemplate;
        this.amqpTemplate.addBeforePublishPostProcessors(new MessagePostProcessorException());
    }


    @Override
    public void afterPropertiesSet() throws Exception {
        // this.amqpTemplate.setCallback(new Consumer<List<? extends com.ziroom.crm.mq.core.data.Message>>() {
        //     @Override
        //     public void accept(List<? extends com.ziroom.crm.mq.core.data.Message> messages) {
        //         for (com.ziroom.crm.mq.core.data.Message message : messages) {
        //             if (message instanceof RabbitMessage) {
        //                 System.out.println("接受到Rabbit send消息 = " + JSON.toJSONString(message));
        //             } else {
        //                 System.out.println("接受到send消息 = " + JSON.toJSONString(message));
        //             }
        //         }
        //     }
        // });
    }

    public String publish(String action, Integer count) {
        if (StringUtils.isBlank(action)) {
            return null;
        }
        //MsgModel msg = new MsgModel(action);
        Msg msg = new Msg(action, random.nextInt(10000) + "", random.nextInt(9999) + "", LocalDateTime.now().toString());
        String msgJson = JSON.toJSONString(msg);

        int finalCount = Objects.isNull(count) ? 1 : count;
        switch (action) {
            case "failOnce":
                sendFail(msgJson);
                break;
            case "failBatch":
                executorService.submit(() -> IntStream.range(0, finalCount).forEach(e -> sendFail(msgJson)));
                break;
            case "failBatchAsync":
                executorService.submit(() -> IntStream.range(0, finalCount).parallel().forEach(e -> sendFail(msgJson)));
                break;
            case "failRandom":
                sendFailRandom(msgJson);
                break;
            case "batchFailRandom":
                IntStream.range(0, finalCount).forEach(e -> sendFailRandom(msgJson));
                break;
            case "batchFailRandomAsync":
                IntStream.range(0, finalCount).parallel().forEach(e -> sendFailRandom(msgJson));
                break;
            default:
                IntStream.range(0, finalCount).parallel().forEach(e -> sendSuccess(msgJson));
        }

        return msgJson + ",success:" + successSet.size() + ",failed:" + failSet.size();
    }


    private void sendSuccess(String msg) {
        MessageProperties messageProperties = new MessageProperties();
        String i = UUID.randomUUID().toString();
        messageProperties.setMessageId(i);
        amqpTemplate.send(Pkg.exchange, Pkg.routing, new Message(msg.getBytes(), messageProperties));
    }

    private void sendFail(String msg) {
        MessagePostProcessor processor = (a) -> {
            return null;
        };
        String i = UUID.randomUUID().toString();
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setHeader(failFlag, "yes");
        messageProperties.setMessageId(i);
        amqpTemplate.send(Pkg.exchange, Pkg.routing, new Message(msg.getBytes(), messageProperties), null);
    }

    private void sendFailRandom(String msg) {
        int i = random.nextInt(10);
        if (i > 3) {
            sendSuccess(msg);
        } else {
            sendFail(msg);
        }
    }


    public static class MessagePostProcessorException implements MessagePostProcessor {

        ThreadLocalRandom random =  ThreadLocalRandom.current();

        /**
         * Change (or replace) the message.
         *
         * @param message the message.
         * @return the message.
         * @throws AmqpException an exception.
         */
        @Override
        public Message postProcessMessage(Message message) throws AmqpException {
            String messageId = message.getMessageProperties().getMessageId();
            set.add(messageId);
            int i = random.nextInt(10);
            if (i > 5 && message.getMessageProperties().getHeaders().containsKey(failFlag)) {
                failSet.add(messageId);
                System.out.println("from error:success=" + successSet.size() + ",fail=" + failSet.size() + ",total=" + set.size());
                throw new AmqpException("send error");
            }
            successSet.add(messageId);
            failSet.remove(messageId);
            System.out.println("from success: success="+successSet.size()+",fail="+failSet.size()+",total="+set.size());
            return message;
        }
    }

}